import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

public class PollerIO extends ServerThread {
    private final Server server;
    //该队列用来接收主线程发来的有事件的连接
    private final Queue<SocketChannel> acceptedQueue;

    public PollerIO(String name, Server server) {
        super(name);
        this.server = server;
        this.acceptedQueue = new LinkedBlockingQueue<>();
    }

    public void addAcceptedConnection(SocketChannel socketChannel) {
        this.acceptedQueue.offer(socketChannel);
        //当子线程拿到到连接后，唤醒selector，让selector继续处理其他连接
        wakeupSelector();
    }

    @Override
    public void run() {
        while (!server.stopped){
            doSelect();
            doAcceptedConnections();
        }
        closeSelector();
    }

    /**
     * 处理acceptedQueue中的连接，注册到selector中
     */
    private void doAcceptedConnections() {
        SocketChannel socketChannel;
        while (!server.stopped && (socketChannel = acceptedQueue.poll()) != null) {
            try {
                socketChannel.register(this.selector, SelectionKey.OP_READ);
            } catch (ClosedChannelException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private void doSelect() {
        try {
            this.selector.select();
            Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey key = iterator.next();
                iterator.remove();
                if (!key.isValid()) {
                    continue;
                }
                if(key.isReadable()||key.isWritable()){
                    handleIO(key);
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }


    }

    private void handleIO(SelectionKey key) {
        if (key.isReadable()) {
            SocketChannel socketChannel = (SocketChannel)key.channel();
            ByteBuffer readBuffer = ByteBuffer.allocate(256);
            try {
                int read = socketChannel.read(readBuffer);
                if (read > 0) {
                    //将数据封装成任务放到线程池
                    BuTask buTask = new BuTask(socketChannel, readBuffer);
                    server.addBusTask(buTask);
                }
            } catch (IOException e) {
                //如果读数据失败，可能是客户端关闭了，这里要关闭客户端
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
            }


        }

    }
}
